package mr

import (
	"context"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"path/filepath"
	"sort"
	"time"

)

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

// Map functions return a slice of KeyValue.
type KeyValue struct {
	Key   string
	Value string
}

// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}
var	workID = 0
var	taskID = -1
var	taskState = 0
// main/mrworker.go calls this function.
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()
	res := GetWorkerID()
	if res == nil {
		return
	} else {
		workID = res.WorkerID
	}
	// fmt.Printf("wokerid: %d\n", res.WorkerID)


	// 引入心跳检查
	ctx, cancel := context.WithCancel(context.Background())
	go heartbeatCheack(ctx, cancel)


	for {
        select {
			case <-ctx.Done():
				// fmt.Println("Worker is exiting...")
				return
			default:
				// 执行任务逻辑
				workerFunc(mapf, reducef, cancel)
        }
	}

}

func doMapTask(fileName string, workID int, taskID int, nReduce int, mapf func(string, string) []KeyValue) bool {
	// fmt.Printf("taskID:%d workerID: %d\n", taskID, workID)
	filePath := fileName
	file, err := os.Open(filePath)
	if err != nil {
		log.Fatalf("cannot open %v", filePath)
		return false
	}
	defer file.Close()
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", filePath)
		return false
	}
	intermediate := []KeyValue{}
	kva := mapf(filePath, string(content))
	intermediate = append(intermediate, kva...)
	//遍历intermediate，按照key hash值 对 nReduce取模存储到对应文件里

	outputDir := "tmp1"
	if err := os.MkdirAll(outputDir, os.ModePerm); err != nil {
		fmt.Printf("could not create output directory: %s", err)
		return false
	}
	// 创建 n 个输出文件的句柄
	outputFiles := make([]*os.File, nReduce)
	for i := 0; i < nReduce; i++ {

		filePath := filepath.Join(outputDir, fmt.Sprintf("mr-%d-%d", taskID, i))
		file, _ := os.Create(filePath)
		outputFiles[i] = file
		defer file.Close() // 程序结束时关闭文件
	}

	for i := 0; i < len(intermediate); i++ {
		key := intermediate[i].Key
		num := ihash(key) % nReduce
		fmt.Fprintf(outputFiles[num], "%v %v\n", intermediate[i].Key, intermediate[i].Value)
	}

	return true
}

func doReduceTask(taskID int, reducef func(string, []string) string) bool {

	//把文件列表统一写入一个文件
	files, _ := filepath.Glob(fmt.Sprintf("./tmp1/*%d", taskID))
	//对文件内容进行多路归并排序
	intermediate := []KeyValue{}
	for _, filename := range files {
		// fmt.Println(filename)
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("cannot open %v", filename)
		}
		for {
			var k string
			var v string
			// 格式化读取文件中的每行内容
			_, err := fmt.Fscanf(file, "%s %s\n", &k, &v)
			// fmt.Printf("%s %s\n", k, v)
			if err != nil {
				break // 文件读取完成或遇到错误时退出
			}
			intermediate = append(intermediate, KeyValue{Key: k, Value: v})
		}

	}
	sort.Sort(ByKey(intermediate))

	// fmt.Println(len(intermediate))

	oname := fmt.Sprintf("mr-out-%d", taskID)
	ofile, _ := os.Create(oname)

	//
	// call Reduce on each distinct key in intermediate[],
	// and print the result to mr-out-0.
	//
	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

		i = j
	}

	ofile.Close()
	return true
}

//心跳检查
func heartbeatCheack(ctx context.Context, cancel context.CancelFunc) {
	ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
	for {
		select {
			case <-ticker.C:
				res := CallHeatbeatCheck(workID)
				if res == nil || res.CoordinatorState != 1 {
					// fmt.Println("Coordinator is unreachable. Exiting...")
					cancel() // 取消上下文，通知其他 goroutine 退出
					return
				}
			case <-ctx.Done():
				return // 退出心跳监测
        }
		
	}
}

// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 100

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	// the "Coordinator.Example" tells the
	// receiving server that we'd like to call
	// the Example() method of struct Coordinator.
	ok := call("Coordinator.Example", &args, &reply)
	if ok {
		// reply.Y should be 100.
		fmt.Printf("reply.Y %v\n", reply.Y)
	} else {
		fmt.Printf("call failed!\n")
	}
}

func CallTask(workerID int, taskState int, taskID int) *GetTaskRes {
	req := GetTaskParam{WorkerID: workerID, TaskState: taskState, TaskID: taskID}

	reply := GetTaskRes{}

	ok := call("Coordinator.GetTask", &req, &reply)
	if ok {
		return &reply
	} else {
		return nil
	}
}

func CallHeatbeatCheck(workID int) *HeatbeatResp {
	req := HeatbeatReq{WorkerID: workID}
	resp := HeatbeatResp{}
	ok := call("Coordinator.CallHeatbeatCheck", &req, &resp)
	if ok {
		return &resp
	} else {
		return nil
	}
}

func GetWorkerID() *GetWorkIDResp {
	req := GetWorkIDReq{}
	resp := GetWorkIDResp{}
	ok := call("Coordinator.GetWorkerID", &req, &resp)
	if ok {
		return &resp
	} else {
		return nil
	}
}

func workerFunc(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string, cancel context.CancelFunc) {
	res := CallTask(workID, taskState, taskID)
	if res == nil {
		cancel()
		return
	}
	taskID = res.TaskID
	// workID = res.WorkerID
	// fmt.Printf("worker接受的数据: %+v\n", res)
	taskState = 0

	if res.TaskType == 1 {
		ok := doMapTask(res.Files[res.TaskID],workID, res.TaskID, res.NReduce, mapf)
		if ok {
			taskState = 2
		} else {
			taskState = 3
		}
	} else if res.TaskType == 2 {
		ok := doReduceTask(res.TaskID, reducef)
		if ok {
			taskState = 2
		} else {
			taskState = 3
		}
	} else if res.TaskType == 3 {
		time.Sleep(time.Duration(500) * time.Millisecond)
		// taskState = 2
	}
}

// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		return false
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	return err == nil
	// fmt.Println(err)
}
